package com.sunday.common.mq.kafka.study.spring.e5_Replying_Kafka_Template;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.IdGenerator;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class MQApi {

    @Autowired
    private ReplyingKafkaTemplate replyingKafkaTemplate;

    @GetMapping("/send/{msg}")
    public void send(@PathVariable String msg) throws ExecutionException, InterruptedException, TimeoutException {
        IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("kRequests", msg);
        producerRecord.headers().add(KafkaHeaders.CORRELATION_ID, defaultIdGenerator.generateId().toString().getBytes(StandardCharsets.UTF_8));
        /**
         * 模板会尝试从配置的回复容器中检测回复主题或分区。 如果容器配置为侦听单个主题或单个 ，则用于设置回复标头。
         * 如果以其他方式配置容器，则用户必须设置回复标头。 在这种情况下，将在初始化期间写入日志消息。
         * headers.add(new RecordHeader(this.replyTopicHeaderName, this.replyTopic));
         */
        RequestReplyFuture<String, String, String> replyFuture = replyingKafkaTemplate.sendAndReceive(producerRecord);
        SendResult<String, String> sendResult = replyFuture.getSendFuture().get(3, TimeUnit.SECONDS);
        log.info("======== {}", sendResult);
        log.info("Sent ok: {}", sendResult.getRecordMetadata());
        /**
         * 从版本 2.6.7 开始，除了检测 s 之外，模板还将调用该函数（如果提供）。
         * 如果它返回异常，则 future 将异常完成。DeserializationExceptionreplyErrorChecker
         */
        try {
            ConsumerRecord<String, String> consumerRecord = replyFuture.get(3, TimeUnit.SECONDS);
            log.info("======== {}", consumerRecord);
            log.info("Return value: {}", consumerRecord.value());
        } catch (Exception e) {
            log.error(e.toString(), e);
        }

    }

    @KafkaListener(id = "server", topics = "kRequests")
    @SendTo
    public String listen(String in) throws InterruptedException {
        log.info("receive : {}", in);
//        TimeUnit.SECONDS.sleep(4);
        return in.toUpperCase();
    }


}
